package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	cluster "github.com/bsm/sarama-cluster"
	"log"
)

// trap SIGINT to trigger a shutdown.
var (
	quitSignal = make(chan bool, 1)
	WorkerLine = make(chan *sarama.ConsumerMessage, 1)
)

func StartKafkaConsumer(topics []string) {
	go func() {
		startKafkaConsumer(topics)
	}()
}

var consumer *cluster.Consumer
var consumerGroupName = "consumerGroup1"
var broker = []string{"127.0.0.1:9092"}

func startKafkaConsumer(topics []string) {

	// init (custom) config, enable errors and notifications
	config := cluster.NewConfig()
	config.Consumer.Return.Errors = true
	config.Consumer.Offsets.Initial = sarama.OffsetNewest
	config.Group.Return.Notifications = true

	// init consumer
	brokers := broker
	consumerGroup := consumerGroupName
	log.Printf("startKafkaConsumer：brokers:%v;consumerGroup:%v;topics:%v\n", brokers, consumerGroup, topics)
	consumerTmp, err := cluster.NewConsumer(brokers, consumerGroup, topics, config)
	log.Printf("startKafkaConsumerEnd：consumerTmp:%+v;\n", consumerTmp)
	consumer = consumerTmp
	if consumer == nil || err != nil {
		log.Printf("StartKafkaConsumer:%v", map[string]interface{}{})
		return
	}

	defer consumer.Close()

	startWorkers(4)
	// consume messages, watch signals
	for {
		select {
		case msg, ok := <-consumer.Messages():
			if ok {
				WorkerLine <- msg
				log.Printf("KafkaConsumerReceiveMsg：topic:%v,value:%v", msg.Topic, string(msg.Value))
				consumer.MarkOffset(msg, "") // mark message as processed
			}
		case <-quitSignal:
			fmt.Println("main func exiting ,begin ShutDown Kafka")
			log.Printf("receive quit signal, stop consume...")
			return
		case ntf := <-consumer.Notifications():
			log.Printf("Rebalanced: %+v\n", ntf)
		case err := <-consumer.Errors():
			log.Printf("kafka consumer report an error%v", map[string]interface{}{"err": err})
			log.Printf("Error: %s\n", err.Error())
		}
	}
}

var consumerController = map[string]func(*sarama.ConsumerMessage){}

func AddController(topicKey string, f func(*sarama.ConsumerMessage)) {
	consumerController[topicKey] = f
}

func startWorkers(workNum int) {
	for i := 0; i < workNum; i++ {
		go func() {
			for msg := range WorkerLine {
				if f, ok := consumerController[msg.Topic]; ok {
					f(msg)
				}
			}
		}()
	}
}

func ShutDownKafkaConsumer() {
	if consumer == nil {
		return
	}
	quitSignal <- true
}
